package com.example.flinktest.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 进行单词频次统计基本思路：先逐行读入文件数据，然后将每一行文字拆分成单
 * 词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次
 */
public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //从文件读取数据，执行读取（存储的元素就是每行的文本）
        DataSource lineDS = env.readTextFile("input/words.txt");
        //进行分词转换数据为（word,count)形式的二元组
        FlatMapOperator<String,Tuple2<String,Long>> wordAndOne = (FlatMapOperator<String, Tuple2<String, Long>>) lineDS.flatMap(
                new FlatMapFunction() {
                    @Override
                    public void flatMap(Object line, Collector out) throws Exception {
                        String[] words = line.toString().split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                }).returns(Types.TUPLE(Types.STRING,Types.LONG));
        //当Lambda表达式使用Java泛型的时候，由于泛型擦除的存在，需要显示的声明类型信息
        // 按照word进行分组，不能使用分组选择器，只能采用位置索引或属性名称进行分组
        UnsortedGrouping<Tuple2<String,Long>> wordAndOneUG = wordAndOne.groupBy(0);
        //分组内聚合统计，进行聚合时同样只能指定聚合字段的位置索引或属性名称
        AggregateOperator<Tuple2<String,Long>> sum = wordAndOneUG.sum(1);
        //打印结果
        sum.print();
    }
}
